/*-------------------------------------------------------------------------
 *
 * pg_subscription.c
 *		复制订阅
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *		src/backend/catalog/pg_subscription.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/genam.h"
#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "catalog/indexing.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "storage/lmgr.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/lsyscache.h"
#include "utils/pg_lsn.h"
#include "utils/rel.h"
#include "utils/syscache.h"

static List *fc_textarray_to_stringlist(ArrayType *fc_textarray);

/*
 * 从系统缓存中获取订阅。
 */
Subscription *
GetSubscription(Oid fc_subid, bool fc_missing_ok)
{
	HeapTuple	fc_tup;
	Subscription *fc_sub;
	Form_pg_subscription fc_subform;
	Datum		fc_datum;
	bool		fc_isnull;

	fc_tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(fc_subid));

	if (!HeapTupleIsValid(fc_tup))
	{
		if (fc_missing_ok)
			return NULL;

		elog(ERROR, "cache lookup failed for subscription %u", fc_subid);
	}

	fc_subform = (Form_pg_subscription) GETSTRUCT(fc_tup);

	fc_sub = (Subscription *) palloc(sizeof(Subscription));
	fc_sub->oid = fc_subid;
	fc_sub->dbid = fc_subform->subdbid;
	fc_sub->skiplsn = fc_subform->subskiplsn;
	fc_sub->name = pstrdup(NameStr(fc_subform->subname));
	fc_sub->owner = fc_subform->subowner;
	fc_sub->enabled = fc_subform->subenabled;
	fc_sub->binary = fc_subform->subbinary;
	fc_sub->stream = fc_subform->substream;
	fc_sub->twophasestate = fc_subform->subtwophasestate;
	fc_sub->disableonerr = fc_subform->subdisableonerr;

	/* 获取 conninfo */
	fc_datum = SysCacheGetAttr(SUBSCRIPTIONOID,
							fc_tup,
							Anum_pg_subscription_subconninfo,
							&fc_isnull);
	Assert(!fc_isnull);
	fc_sub->conninfo = TextDatumGetCString(fc_datum);

	/* 获取 slotname */
	fc_datum = SysCacheGetAttr(SUBSCRIPTIONOID,
							fc_tup,
							Anum_pg_subscription_subslotname,
							&fc_isnull);
	if (!fc_isnull)
		fc_sub->slotname = pstrdup(NameStr(*DatumGetName(fc_datum)));
	else
		fc_sub->slotname = NULL;

	/* 获取synccommit */
	fc_datum = SysCacheGetAttr(SUBSCRIPTIONOID,
							fc_tup,
							Anum_pg_subscription_subsynccommit,
							&fc_isnull);
	Assert(!fc_isnull);
	fc_sub->synccommit = TextDatumGetCString(fc_datum);

	/* 获取出版物 */
	fc_datum = SysCacheGetAttr(SUBSCRIPTIONOID,
							fc_tup,
							Anum_pg_subscription_subpublications,
							&fc_isnull);
	Assert(!fc_isnull);
	fc_sub->publications = fc_textarray_to_stringlist(DatumGetArrayTypeP(fc_datum));

	ReleaseSysCache(fc_tup);

	return fc_sub;
}

/*
 * 返回在给定数据库中定义的订阅数量。
 * 由dropdb()使用以检查数据库是否确实可以被删除。
 */
int CountDBSubscriptions(Oid fc_dbid)
{
	int			fc_nsubs = 0;
	Relation	fc_rel;
	ScanKeyData fc_scankey;
	SysScanDesc fc_scan;
	HeapTuple	fc_tup;

	fc_rel = table_open(SubscriptionRelationId, RowExclusiveLock);

	ScanKeyInit(&fc_scankey,
				Anum_pg_subscription_subdbid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_dbid));

	fc_scan = systable_beginscan(fc_rel, InvalidOid, false,
							  NULL, 1, &fc_scankey);

	while (HeapTupleIsValid(fc_tup = systable_getnext(fc_scan)))
		fc_nsubs++;

	systable_endscan(fc_scan);

	table_close(fc_rel, NoLock);

	return fc_nsubs;
}

/*
 * 释放由订阅结构分配的内存。
 */
void FreeSubscription(Subscription *fc_sub)
{
	pfree(fc_sub->name);
	pfree(fc_sub->conninfo);
	if (fc_sub->slotname)
		pfree(fc_sub->slotname);
	list_free_deep(fc_sub->publications);
	pfree(fc_sub);
}

/*
 * 禁用给定的订阅。
 */
void DisableSubscription(Oid fc_subid)
{
	Relation	fc_rel;
	bool		fc_nulls[Natts_pg_subscription];
	bool		fc_replaces[Natts_pg_subscription];
	Datum		fc_values[Natts_pg_subscription];
	HeapTuple	fc_tup;

	/* 在目录中查找订阅 */
	fc_rel = table_open(SubscriptionRelationId, RowExclusiveLock);
	fc_tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(fc_subid));

	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR, "cache lookup failed for subscription %u", fc_subid);

	LockSharedObject(SubscriptionRelationId, fc_subid, 0, AccessShareLock);

	/* 形成一个新元组。 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));
	memset(fc_replaces, false, sizeof(fc_replaces));

	/* 将订阅设置为禁用。 */
	fc_values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(false);
	fc_replaces[Anum_pg_subscription_subenabled - 1] = true;

	/* 更新目录 */
	fc_tup = heap_modify_tuple(fc_tup, RelationGetDescr(fc_rel), fc_values, fc_nulls,
							fc_replaces);
	CatalogTupleUpdate(fc_rel, &fc_tup->t_self, fc_tup);
	heap_freetuple(fc_tup);

	table_close(fc_rel, NoLock);
}

/*
 * get_subscription_oid - 给定订阅名称，查找OID
 *
 * 如果missing_ok为false，则在未找到名称时抛出错误。如果为true，则
 * 仅返回InvalidOid。
 */
Oid get_subscription_oid(const char *fc_subname, bool fc_missing_ok)
{
	Oid			fc_oid;

	fc_oid = GetSysCacheOid2(SUBSCRIPTIONNAME, Anum_pg_subscription_oid,
						  MyDatabaseId, CStringGetDatum(fc_subname));
	if (!OidIsValid(fc_oid) && !fc_missing_ok)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("subscription \"%s\" does not exist", fc_subname)));
	return fc_oid;
}

/*
 * get_subscription_name - 给定订阅OID，查找名称
 *
 * 如果missing_ok为false，则在未找到名称时抛出错误。如果为true，则
 * 仅返回NULL。
 */
char * get_subscription_name(Oid fc_subid, bool fc_missing_ok)
{
	HeapTuple	fc_tup;
	char	   *fc_subname;
	Form_pg_subscription fc_subform;

	fc_tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(fc_subid));

	if (!HeapTupleIsValid(fc_tup))
	{
		if (!fc_missing_ok)
			elog(ERROR, "cache lookup failed for subscription %u", fc_subid);
		return NULL;
	}

	fc_subform = (Form_pg_subscription) GETSTRUCT(fc_tup);
	fc_subname = pstrdup(NameStr(fc_subform->subname));

	ReleaseSysCache(fc_tup);

	return fc_subname;
}

/*
 * 将文本数组转换为字符串列表。
 *
 * 注意：生成的字符串列表在这里进行内存分配。
 */
static List * fc_textarray_to_stringlist(ArrayType *fc_textarray)
{
	Datum	   *fc_elems;
	int			fc_nelems,
				fc_i;
	List	   *fc_res = NIL;

	deconstruct_array(fc_textarray,
					  TEXTOID, -1, false, TYPALIGN_INT,
					  &fc_elems, NULL, &fc_nelems);

	if (fc_nelems == 0)
		return NIL;

	for (fc_i = 0; fc_i < fc_nelems; fc_i++)
		fc_res = lappend(fc_res, makeString(TextDatumGetCString(fc_elems[fc_i])));

	return fc_res;
}

/*
 * 为订阅表添加新的状态记录。
 */
void AddSubscriptionRelState(Oid fc_subid, Oid fc_relid, char fc_state,
						XLogRecPtr fc_sublsn)
{
	Relation	fc_rel;
	HeapTuple	fc_tup;
	bool		fc_nulls[Natts_pg_subscription_rel];
	Datum		fc_values[Natts_pg_subscription_rel];

	LockSharedObject(SubscriptionRelationId, fc_subid, 0, AccessShareLock);

	fc_rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);

	/* 尝试寻找现有映射。 */
	fc_tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
							  ObjectIdGetDatum(fc_relid),
							  ObjectIdGetDatum(fc_subid));
	if (HeapTupleIsValid(fc_tup))
		elog(ERROR, "subscription table %u in subscription %u already exists",
			 fc_relid, fc_subid);

	/* 形成元组。 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));
	fc_values[Anum_pg_subscription_rel_srsubid - 1] = ObjectIdGetDatum(fc_subid);
	fc_values[Anum_pg_subscription_rel_srrelid - 1] = ObjectIdGetDatum(fc_relid);
	fc_values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(fc_state);
	if (fc_sublsn != InvalidXLogRecPtr)
		fc_values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(fc_sublsn);
	else
		fc_nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;

	fc_tup = heap_form_tuple(RelationGetDescr(fc_rel), fc_values, fc_nulls);

	/* 将元组插入目录。 */
	CatalogTupleInsert(fc_rel, fc_tup);

	heap_freetuple(fc_tup);

	/* 清理。 */
	table_close(fc_rel, NoLock);
}

/*
 * 更新订阅表的状态。
 */
void UpdateSubscriptionRelState(Oid fc_subid, Oid fc_relid, char fc_state,
						   XLogRecPtr fc_sublsn)
{
	Relation	fc_rel;
	HeapTuple	fc_tup;
	bool		fc_nulls[Natts_pg_subscription_rel];
	Datum		fc_values[Natts_pg_subscription_rel];
	bool		fc_replaces[Natts_pg_subscription_rel];

	LockSharedObject(SubscriptionRelationId, fc_subid, 0, AccessShareLock);

	fc_rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);

	/* 尝试寻找现有映射。 */
	fc_tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
							  ObjectIdGetDatum(fc_relid),
							  ObjectIdGetDatum(fc_subid));
	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR, "subscription table %u in subscription %u does not exist",
			 fc_relid, fc_subid);

	/* 更新元组。 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));
	memset(fc_replaces, false, sizeof(fc_replaces));

	fc_replaces[Anum_pg_subscription_rel_srsubstate - 1] = true;
	fc_values[Anum_pg_subscription_rel_srsubstate - 1] = CharGetDatum(fc_state);

	fc_replaces[Anum_pg_subscription_rel_srsublsn - 1] = true;
	if (fc_sublsn != InvalidXLogRecPtr)
		fc_values[Anum_pg_subscription_rel_srsublsn - 1] = LSNGetDatum(fc_sublsn);
	else
		fc_nulls[Anum_pg_subscription_rel_srsublsn - 1] = true;

	fc_tup = heap_modify_tuple(fc_tup, RelationGetDescr(fc_rel), fc_values, fc_nulls,
							fc_replaces);

	/* 更新目录。 */
	CatalogTupleUpdate(fc_rel, &fc_tup->t_self, fc_tup);

	/* 清理。 */
	table_close(fc_rel, NoLock);
}

/*
 * 获取订阅表的状态。
 *
 * 当表不在订阅中时返回SUBREL_STATE_UNKNOWN。
 */
char GetSubscriptionRelState(Oid fc_subid, Oid fc_relid, XLogRecPtr *fc_sublsn)
{
	HeapTuple	fc_tup;
	char		fc_substate;
	bool		fc_isnull;
	Datum		fc_d;
	Relation	fc_rel;

	/*
	 * 这是为了避免与尝试删除此relstate的AlterSubscription的竞争条件。
	 */
	fc_rel = table_open(SubscriptionRelRelationId, AccessShareLock);

	/* 尝试寻找映射。 */
	fc_tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
						  ObjectIdGetDatum(fc_relid),
						  ObjectIdGetDatum(fc_subid));

	if (!HeapTupleIsValid(fc_tup))
	{
		table_close(fc_rel, AccessShareLock);
		*fc_sublsn = InvalidXLogRecPtr;
		return SUBREL_STATE_UNKNOWN;
	}

	/* 获取状态。 */
	fc_substate = ((Form_pg_subscription_rel) GETSTRUCT(fc_tup))->srsubstate;

	/* 获取LSN */
	fc_d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, fc_tup,
						Anum_pg_subscription_rel_srsublsn, &fc_isnull);
	if (fc_isnull)
		*fc_sublsn = InvalidXLogRecPtr;
	else
		*fc_sublsn = DatumGetLSN(fc_d);

	/* 清理 */
	ReleaseSysCache(fc_tup);

	table_close(fc_rel, AccessShareLock);

	return fc_substate;
}

/*
 * 删除订阅关系映射。这些可以是特定
 * 订阅的，或者是特定关系的，或者两者都有。
 */
void RemoveSubscriptionRel(Oid fc_subid, Oid fc_relid)
{
	Relation	fc_rel;
	TableScanDesc fc_scan;
	ScanKeyData fc_skey[2];
	HeapTuple	fc_tup;
	int			fc_nkeys = 0;

	fc_rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);

	if (OidIsValid(fc_subid))
	{
		ScanKeyInit(&fc_skey[fc_nkeys++],
					Anum_pg_subscription_rel_srsubid,
					BTEqualStrategyNumber,
					F_OIDEQ,
					ObjectIdGetDatum(fc_subid));
	}

	if (OidIsValid(fc_relid))
	{
		ScanKeyInit(&fc_skey[fc_nkeys++],
					Anum_pg_subscription_rel_srrelid,
					BTEqualStrategyNumber,
					F_OIDEQ,
					ObjectIdGetDatum(fc_relid));
	}

	/* 进行搜索并删除我们找到的内容。 */
	fc_scan = table_beginscan_catalog(fc_rel, fc_nkeys, fc_skey);
	while (HeapTupleIsValid(fc_tup = heap_getnext(fc_scan, ForwardScanDirection)))
	{
		Form_pg_subscription_rel fc_subrel;

		fc_subrel = (Form_pg_subscription_rel) GETSTRUCT(fc_tup);

		/*
		 * 我们不允许在表同步进行时删除关系映射，除非调用者同时更新对应的
		 * 订阅。这是为了确保在相应的表被删除时，不会在系统中留下表同步插槽或来源。
		 */
		if (!OidIsValid(fc_subid) && fc_subrel->srsubstate != SUBREL_STATE_READY)
		{
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("could not drop relation mapping for subscription \"%s\"",
							get_subscription_name(fc_subrel->srsubid, false)),
					 errdetail("Table synchronization for relation \"%s\" is in progress and is in state \"%c\".",
							   get_rel_name(fc_relid), fc_subrel->srsubstate),

			/*
			 * 翻译者：第一个 %s 是 SQL ALTER 命令，第二个 %s 是
			 * SQL DROP 命令
			 */
					 errhint("Use %s to enable subscription if not already enabled or use %s to drop the subscription.",
							 "ALTER SUBSCRIPTION ... ENABLE",
							 "DROP SUBSCRIPTION ...")));
		}

		CatalogTupleDelete(fc_rel, &fc_tup->t_self);
	}
	table_endscan(fc_scan);

	table_close(fc_rel, RowExclusiveLock);
}

/*
 * 该订阅是否有任何关系？
 *
 * 仅在您需要知道真假时使用此函数，并且在
 * 您不需要GetSubscriptionRelations返回的列表时。
 */
bool HasSubscriptionRelations(Oid fc_subid)
{
	Relation	fc_rel;
	ScanKeyData fc_skey[1];
	SysScanDesc fc_scan;
	bool		fc_has_subrels;

	fc_rel = table_open(SubscriptionRelRelationId, AccessShareLock);

	ScanKeyInit(&fc_skey[0],
				Anum_pg_subscription_rel_srsubid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_subid));

	fc_scan = systable_beginscan(fc_rel, InvalidOid, false,
							  NULL, 1, fc_skey);

	/* 如果至少存在一个元组，那么订阅就有表。 */
	fc_has_subrels = HeapTupleIsValid(systable_getnext(fc_scan));

	/* 清理 */
	systable_endscan(fc_scan);
	table_close(fc_rel, AccessShareLock);

	return fc_has_subrels;
}

/*
 * 获取订阅的所有关系。
 *
 * 返回的列表是在当前内存上下文中分配的。
 */
List * GetSubscriptionRelations(Oid fc_subid)
{
	List	   *fc_res = NIL;
	Relation	fc_rel;
	HeapTuple	fc_tup;
	ScanKeyData fc_skey[1];
	SysScanDesc fc_scan;

	fc_rel = table_open(SubscriptionRelRelationId, AccessShareLock);

	ScanKeyInit(&fc_skey[0],
				Anum_pg_subscription_rel_srsubid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_subid));

	fc_scan = systable_beginscan(fc_rel, InvalidOid, false,
							  NULL, 1, fc_skey);

	while (HeapTupleIsValid(fc_tup = systable_getnext(fc_scan)))
	{
		Form_pg_subscription_rel fc_subrel;
		SubscriptionRelState *fc_relstate;
		Datum		fc_d;
		bool		fc_isnull;

		fc_subrel = (Form_pg_subscription_rel) GETSTRUCT(fc_tup);

		fc_relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
		fc_relstate->relid = fc_subrel->srrelid;
		fc_relstate->state = fc_subrel->srsubstate;
		fc_d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, fc_tup,
							Anum_pg_subscription_rel_srsublsn, &fc_isnull);
		if (fc_isnull)
			fc_relstate->lsn = InvalidXLogRecPtr;
		else
			fc_relstate->lsn = DatumGetLSN(fc_d);

		fc_res = lappend(fc_res, fc_relstate);
	}

	/* 清理 */
	systable_endscan(fc_scan);
	table_close(fc_rel, AccessShareLock);

	return fc_res;
}

/*
 * 获取订阅中不处于准备状态的所有关系。
 *
 * 返回的列表是在当前内存上下文中分配的。
 */
List * GetSubscriptionNotReadyRelations(Oid fc_subid)
{
	List	   *fc_res = NIL;
	Relation	fc_rel;
	HeapTuple	fc_tup;
	int			fc_nkeys = 0;
	ScanKeyData fc_skey[2];
	SysScanDesc fc_scan;

	fc_rel = table_open(SubscriptionRelRelationId, AccessShareLock);

	ScanKeyInit(&fc_skey[fc_nkeys++],
				Anum_pg_subscription_rel_srsubid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_subid));

	ScanKeyInit(&fc_skey[fc_nkeys++],
				Anum_pg_subscription_rel_srsubstate,
				BTEqualStrategyNumber, F_CHARNE,
				CharGetDatum(SUBREL_STATE_READY));

	fc_scan = systable_beginscan(fc_rel, InvalidOid, false,
							  NULL, fc_nkeys, fc_skey);

	while (HeapTupleIsValid(fc_tup = systable_getnext(fc_scan)))
	{
		Form_pg_subscription_rel fc_subrel;
		SubscriptionRelState *fc_relstate;
		Datum		fc_d;
		bool		fc_isnull;

		fc_subrel = (Form_pg_subscription_rel) GETSTRUCT(fc_tup);

		fc_relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState));
		fc_relstate->relid = fc_subrel->srrelid;
		fc_relstate->state = fc_subrel->srsubstate;
		fc_d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, fc_tup,
							Anum_pg_subscription_rel_srsublsn, &fc_isnull);
		if (fc_isnull)
			fc_relstate->lsn = InvalidXLogRecPtr;
		else
			fc_relstate->lsn = DatumGetLSN(fc_d);

		fc_res = lappend(fc_res, fc_relstate);
	}

	/* 清理 */
	systable_endscan(fc_scan);
	table_close(fc_rel, AccessShareLock);

	return fc_res;
}
